package com.wen;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import redis.clients.jedis.Jedis;

import java.net.InetSocketAddress;
import java.util.List;

public class WenCanalRedisSyncClient {

    public static void main(String[] args) {
        // 1. 连接Canal服务
        String canalServerIp = "127.0.0.1";
        int canalServerPort = 11111;
        String destination = "example";
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(canalServerIp, canalServerPort),
                destination,
                "", ""
        );

        // 2. 连接Redis
        Jedis jedis = new Jedis("127.0.0.1", 6379);
        System.out.println("Successfully connected to Redis");

        int batchSize = 1000;
        try {
            connector.connect();
            // 订阅所有库的所有表
            connector.subscribe(".*\\..*");
            // 回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的位置开始拿
            connector.rollback();

            while (true) {
                // 3. 从Canal获取数据
                Message message = connector.getWithoutAck(batchSize);
                long batchId = message.getId();
                int size = message.getEntries().size();

                if (batchId == -1 || size == 0) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    // 4. 处理获取到的数据
                    processEntries(message.getEntries(), jedis);
                }

                // 5. 确认处理完成
                connector.ack(batchId);
            }
        } finally {
            connector.disconnect();
            jedis.close();
        }
    }

    private static void processEntries(List<CanalEntry.Entry> entries, Jedis jedis) {
        for (CanalEntry.Entry entry : entries) {
            // 过滤掉非行数据变更的事件
            if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                continue;
            }

            try {
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                CanalEntry.EventType eventType = rowChange.getEventType();
                String tableName = entry.getHeader().getTableName();
                String dbName = entry.getHeader().getSchemaName();
                String redisKeyPrefix = dbName + ":" + tableName + ":";

                System.out.println("================> binlog[" + entry.getHeader().getLogfileName() + ":" + entry.getHeader().getLogfileOffset() + "] , name[" + dbName + "." + tableName + "], eventType : " + eventType);

                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    // 处理删除事件
                    if (eventType == CanalEntry.EventType.DELETE) {
                        String key = redisKeyPrefix + getPrimaryKey(rowData.getBeforeColumnsList());
                        jedis.del(key);
                        System.out.println("Deleted Redis key: " + key);
                    }
                    // 处理插入和更新事件
                    else if (eventType == CanalEntry.EventType.INSERT || eventType == CanalEntry.EventType.UPDATE) {
                        String key = redisKeyPrefix + getPrimaryKey(rowData.getAfterColumnsList());
                        String value = columnsToJson(rowData.getAfterColumnsList());
                        jedis.set(key, value);
                        System.out.println("Set Redis key: " + key + ", Value: " + value);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    // 简单获取主键ID（假设第一列是主键）
    private static String getPrimaryKey(List<CanalEntry.Column> columns) {
        // 更健壮的方式是检查表结构获取主键名，但这里为简化示例，直接取id列
        for (CanalEntry.Column column : columns) {
            if (column.getName().equals("id")) {
                return column.getValue();
            }
        }
        return columns.get(0).getValue();
    }

    // 将列数据转换为JSON字符串
    private static String columnsToJson(List<CanalEntry.Column> columns) {
        JSONObject json = new JSONObject();
        for (CanalEntry.Column column : columns) {
            json.put(column.getName(), column.getValue());
        }
        return json.toJSONString();
    }
}